package com.service;


import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.dto.*;
import com.model.MobileStation;
import com.model.Rover;
import com.repository.MobileStationMapper;
import com.repository.RoverRepository;
import com.util.ElasticSearchUtil;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.*;
import org.elasticsearch.search.aggregations.bucket.filter.ParsedFilter;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
import org.elasticsearch.search.aggregations.bucket.histogram.Histogram;
import org.elasticsearch.search.aggregations.bucket.histogram.ParsedDateHistogram;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedStringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.metrics.cardinality.ParsedCardinality;
import org.elasticsearch.search.aggregations.metrics.max.ParsedMax;
import org.elasticsearch.search.aggregations.metrics.tophits.ParsedTopHits;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.aggregation.AggregatedPage;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.data.elasticsearch.core.query.Query;
import org.springframework.data.elasticsearch.core.query.SearchQuery;
import org.springframework.data.util.CloseableIterator;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.util.*;
import java.util.stream.Collectors;

/**
 * <p>
 * 流动站信息表 服务实现类
 * </p>
 *
 * @author wangjiang
 * @since 2020-11-03
 */
@Service
public class MobileStationServiceImpl  implements IMobileStationService {

    @Autowired
    private ElasticsearchRestTemplate elasticsearchRestTemplate;
    @Autowired
    private MobileStationMapper mobileStationMapper;
@Autowired
private RoverRepository roverRepository;





    /**
     * 流动站名称改成ntripname那个吧
     *
     * 侯先生:
     * 设备名称是username
     *
     * 侯先生:
     * 挂在点mountpoint
     * @param vm
     * @return
     */
    @Override
    public ResponseEntity<IPage<MobileStationPlusDTO>> pagePlus(MobileStationPageVM vm, Integer interval) {
        Page<MobileStation> page = new Page<>(vm.getCurrent(), vm.getSize());
        Page<MobileStationPlusDTO> pageResult = null;
        List<MobileStationPlusDTO> records = null;
        if(CollectionUtils.isEmpty(records)){
            return ResponseEntity.ok(pageResult);
        }

        for (MobileStationPlusDTO record : records) {

            Double lastLoginDate = this.getLastLoginDate(record);
            if(lastLoginDate==null) {
                continue;
            }
            List<Rover.OneRover> searchHits = this.getRoverList(record,lastLoginDate,interval);
            record.setRoverList(searchHits);
            //统计当前在线时长
            record.setOnlineTime(  this.getCurrentOnline(lastLoginDate,record));
            //设置网格点，小区，子网
            if(CollectionUtils.isEmpty(searchHits)){
                continue;
            }

            Rover.OneRover rover = searchHits.get(0);
            if(rover==null){
                continue;
            }
            record.setCellId(rover.getCellid());
            record.setSubId(rover.getSubid());
            if(rover.getVrsid() != null && rover.getVrsid() != 0) {
                record.setNetPointId(rover.getVrsid().toString());
            }



        }
        return ResponseEntity.ok(pageResult);
    }

    /**
     * 统计当前在线时长
     * @param lastLoginDate
     * @param record
     */
    private String getCurrentOnline(Double lastLoginDate, MobileStationPlusDTO record) {
        BoolQueryBuilder boolQueryBuilder = this.getQuery(record);
        NativeSearchQueryBuilder builder = new NativeSearchQueryBuilder();

        builder .withQuery(boolQueryBuilder);
        builder.withPageable(PageRequest.of(1,1));
        AggregationBuilder aggregationBuilder = AggregationBuilders.max("gpggaDate").field("gpggaDate");
        builder.addAggregation((AbstractAggregationBuilder)aggregationBuilder);
//        Query query = builder.build();
//        SearchQuery searchQuery = builder.build();

        SearchQuery query = builder.build();
//        SearchHits<Rover.OneRover> search = elasticsearchRestTemplate.search(query, Rover.OneRover.class);
        AggregatedPage<Rover.OneRover> search = elasticsearchRestTemplate.queryForPage(query, Rover.OneRover.class);
        List<Aggregation> aggregations = search.getAggregations().asList();
        if(CollectionUtils.isEmpty(aggregations)){
            return null;
        }
        ParsedMax parsedMax = (ParsedMax) aggregations.get(0);
        if(parsedMax==null){
            return null;
        }
        Double now =  parsedMax.getValue();
        if(Objects.equals(now.toString(),"-Infinity")){
            return null;
        }
        Double online = now - lastLoginDate;
        Double danwei = Double.valueOf( 1000 * 60 * 60);
        Double time = online/danwei;
        return time.toString();

    }

    /**
     * 获取流动站数据
     * @param record
     * @return
     */
    private  List<Rover.OneRover> getRoverList(MobileStationPlusDTO record,Double lastLoginDate,Integer interval) {
        if(lastLoginDate == null){
            return null;
        }
        NativeSearchQueryBuilder builder = new NativeSearchQueryBuilder();
        BoolQueryBuilder boolQueryBuilder = this.getQuery(record);
        boolQueryBuilder.must(QueryBuilders.rangeQuery("gpggaDate").gte(lastLoginDate));
        builder.withQuery(boolQueryBuilder);
        builder.withSort(SortBuilders.fieldSort("gpggaDate").order(SortOrder.ASC));
        builder.withPageable(PageRequest.of(1,1));
        //分组时间间隔来统计 返回一组数据 必须分组，内存溢出。
        AggregationBuilder agg = AggregationBuilders.dateHistogram("gpggaDate")
                .interval(interval)
                .field("gpggaDate");
        AggregationBuilder agg2 = AggregationBuilders
                .topHits("rover")
                .sort("gpggaDate", SortOrder.DESC)
                .fetchSource(new String[]{"ntripname","mountpiont","username","time","GPGGA","subid","cellid","vrsid","gpggaDate","latitude","longitude","loginDate"},null)
                .size(1);
        agg.subAggregation(agg2);
        builder.addAggregation((AbstractAggregationBuilder) agg);
//                Query query = builder.build();
//
//        SearchHits<Rover.OneRover> search = elasticsearchRestTemplate.search(query, Rover.OneRover.class);
        SearchQuery query = builder.build();
        AggregatedPage<Rover.OneRover> search = elasticsearchRestTemplate.queryForPage(query, Rover.OneRover.class);
        List<Aggregation> aggres = search.getAggregations().asList();
        if(CollectionUtils.isEmpty(aggres)){
            return null;
        }
        ParsedDateHistogram parsedTopHits = (ParsedDateHistogram) aggres.get(0);
        List<? extends Histogram.Bucket> buckets = parsedTopHits.getBuckets();
        if(CollectionUtils.isEmpty(buckets)){
            return null;
        }
        List<Rover.OneRover> roverList = new ArrayList<>();
        for (Histogram.Bucket bucket : buckets) {
            List<Aggregation> aggres3 = bucket.getAggregations().asList();
            if(CollectionUtils.isEmpty(aggres3)){
                continue;
            }

            ParsedTopHits topHits = (ParsedTopHits) aggres3.get(0);
            if(topHits.getHits()==null || topHits.getHits().getHits().length<1){
                continue;
            }
            org.elasticsearch.search.SearchHit hit = topHits.getHits().getHits()[0];
            if(hit==null){
                continue;
            }
            String json = hit.toString();
            JsonRoverDTO jsonRoverDTO = JSONObject.parseObject(json, new TypeReference<JsonRoverDTO>() {});
            Rover.OneRover oneRover = jsonRoverDTO.get_source();

            roverList.add(oneRover);
        }


        return roverList;
    }
    public BoolQueryBuilder getQuery(MobileStationPlusDTO record){
        if(StringUtils.isEmpty(record.getNtripAccount())){
            return null;
        }
        Map<String,String> must = new HashMap<>();
        must.put("ntripname",record.getNtripAccount());
        must.put("username",record.getName());
        must.put("mountpiont",record.getMountPointId());
        BoolQueryBuilder queryBuilder = ElasticSearchUtil.getQueryBuilder(must, null, null, null);
        return queryBuilder;
    }
    /**
     * 获取最后的登陆时间
     * @param
     * @param
     * @return
     */
    private Double getLastLoginDate(MobileStationPlusDTO record ) {
        BoolQueryBuilder boolQueryBuilder = this.getQuery(record);
        NativeSearchQueryBuilder builder = new NativeSearchQueryBuilder();
        builder .withQuery(boolQueryBuilder);
        builder.withPageable(PageRequest.of(1,1));
        //wvx
        AggregationBuilder aggregationBuilder = AggregationBuilders.max("loginDate").field("loginDate");
        builder.addAggregation((AbstractAggregationBuilder)aggregationBuilder);
//        Query query = builder.build();
//        SearchHits<Rover.OneRover> search = elasticsearchRestTemplate.search(query, Rover.OneRover.class);
        SearchQuery query = builder.build();
        AggregatedPage<Rover.OneRover> search = elasticsearchRestTemplate.queryForPage(query, Rover.OneRover.class);
        List<Aggregation> aggregations = search.getAggregations().asList();
        if(CollectionUtils.isEmpty(aggregations)){
            return null;
        }
        ParsedMax parsedMax = (ParsedMax) aggregations.get(0);
        if(parsedMax==null){
            return null;
        }
        Double value = parsedMax.getValue();
        if(Objects.equals(value.toString(),"-Infinity")){
            return null;
        }
        return value;
    }





    @Override
    public List<MobileStationMonitorDTO>  getMobileStationMonitor(String stateStatus, String search,Integer minute) {
        List<MobileStationMonitorDTO> monitorDTOList = mobileStationMapper.getCompanys(stateStatus,search);
      List<MobileStationMonitorDTO.MobileDevice> devices =   mobileStationMapper.getMobileStations(stateStatus,search);

        for (MobileStationMonitorDTO dto : monitorDTOList) {
            List<MobileStationMonitorDTO.MobileDevice> list = devices.stream().filter((item) -> {
                return Objects.equals(item.getCmy(), dto.getCmy());
            }).collect(Collectors.toList());
            //查询流动站的最新坐标
            this.getStationCoordinate(list,minute);
            if(StringUtils.isEmpty(stateStatus)) {
                dto.setDeviceList(list);
                continue;
            }
            //对移动与静止进行筛选
            switch (stateStatus){
                case "move":
                    List<MobileStationMonitorDTO.MobileDevice> move = list.stream().filter(item -> {
                        return Objects.equals(item.getState(), "move");
                    }).collect(Collectors.toList());
                    dto.setDeviceList(move);
                    break;
                case "stop":
                    List<MobileStationMonitorDTO.MobileDevice> stop = list.stream().filter(item -> {
                        return Objects.equals(item.getState(), "stop");
                    }).collect(Collectors.toList());
                    dto.setDeviceList(stop);
                    break;
            }
        }
        //空集合的不要
        List<MobileStationMonitorDTO> collect = monitorDTOList.stream().filter(item -> {
            return !CollectionUtils.isEmpty(item.getDeviceList());
        }).collect(Collectors.toList());
        return collect;
    }

    public BoolQueryBuilder getQuery(MobileStationMonitorDTO.MobileDevice mobileDevice){
        if(StringUtils.isEmpty(mobileDevice.getNtripAccount())){
            return null;
        }
        Map<String,String> must = new HashMap<>();
        must.put("ntripname",mobileDevice.getNtripAccount());
        BoolQueryBuilder queryBuilder = ElasticSearchUtil.getQueryBuilder(must, null, null, null);
        return queryBuilder;
    }

    /**
     * 查询流动站的最新坐标
     * @param mobileDevices
     */
    private void getStationCoordinate(List<MobileStationMonitorDTO.MobileDevice> mobileDevices,Integer minute) {
            if(CollectionUtils.isEmpty(mobileDevices)){
                return;}
        MobileStationMonitorDTO.MobileDevice record = mobileDevices.get(0);
            if(record==null) {
                return;
            }
        Double mobileLatesGpdate = this.getMobileLatesGpdate(record);
            if(mobileLatesGpdate==null){
                return;
            }
            //查询最新时间线的坐标
            NativeSearchQueryBuilder builder = new NativeSearchQueryBuilder();
            BoolQueryBuilder boolQueryBuilder = this.getQuery(record);
            boolQueryBuilder.must(QueryBuilders.matchQuery("gpggaDate",mobileLatesGpdate));
            builder.withQuery(boolQueryBuilder);
            //分组过滤查询是移动还是静止
            Double preTen = mobileLatesGpdate-(minute*60*1000);//10分钟前的时间

            AggregationBuilder aggregationBuilder =
                    AggregationBuilders.filter("gpdate", QueryBuilders.rangeQuery("gpggaDate").gte(preTen));
            AggregationBuilder aggTwo = AggregationBuilders.terms("username").field("username");
            aggregationBuilder.subAggregation(aggTwo);
            AggregationBuilder aggThree = AggregationBuilders.cardinality("latitude").field("latitude");
             AggregationBuilder aggFour = AggregationBuilders.cardinality("longitude").field("longitude");
             aggTwo.subAggregation(aggThree);
             aggTwo.subAggregation(aggFour);
            builder.addAggregation((AbstractAggregationBuilder)aggregationBuilder);

//            Query query = builder.build();
//            SearchHits<Rover.OneRover> search = elasticsearchRestTemplate.search(query, Rover.OneRover.class);
        SearchQuery query = builder.build();
        AggregatedPage<Rover.OneRover> search = elasticsearchRestTemplate.queryForPage(query, Rover.OneRover.class);
            if(search==null){
               return;
            }
//            List<SearchHit<Rover.OneRover>> searchHits = search.getSearchHits();
//            if(CollectionUtils.isEmpty(searchHits)){
//                return;
//            }
            Map<String , Rover.OneRover> map = new HashMap<>();
            //获取最新坐标
//        for (SearchHit<Rover.OneRover> searchHit : searchHits) {
//            Rover.OneRover content = searchHit.getContent();
//            if(content==null){
//                continue;
//            }
//            map.put(content.getUsername(),content);
//        }

        //获取是否移动
        Aggregations aggregations = search.getAggregations();
        Aggregation gpdate = aggregations.get("gpdate");
        Map<String, Object> metaData = gpdate.getMetaData();
        ParsedFilter parsedFilter = (ParsedFilter) gpdate;
        Aggregations agg2 = parsedFilter.getAggregations();
        Aggregation username = agg2.get("username");
        ParsedStringTerms parsedStringTerms = (ParsedStringTerms)username;
        List<? extends Terms.Bucket> buckets = parsedStringTerms.getBuckets();
        Map<String,Boolean> moveMap = new HashMap<>();
        for (Terms.Bucket bucket : buckets) {
            String key = (String)bucket.getKey();
            Boolean flag = false;
            Aggregations agg = bucket.getAggregations();
            for (Aggregation aggregation : agg) {
                ParsedCardinality parsedCardinality = (ParsedCardinality)aggregation;
                long value = parsedCardinality.getValue();
                if(value>1){
                    flag = true;
                }

            }
            moveMap.put(key,flag);

        }
        //录入坐标
        //录入移动禁止
        for (MobileStationMonitorDTO.MobileDevice mobileDevice : mobileDevices) {
            if(mobileDevice==null){
                continue;
            }
            Boolean aBoolean = moveMap.get(mobileDevice.getName());

            mobileDevice.setState(aBoolean == null ? "stop" : aBoolean ? "move" : "stop");
            Rover.OneRover oneRover = map.get(mobileDevice.getName());
            if(oneRover==null){
                continue;
            }
            mobileDevice.setLongitude(oneRover.getLongitude());
            mobileDevice.setLatitude(oneRover.getLatitude());

        }

    }

    /**
     * 获取 流动站的最新发送时间
     * @param mobileDevice
     * @return
     */public Double getMobileLatesGpdate(MobileStationMonitorDTO.MobileDevice mobileDevice){



        BoolQueryBuilder boolQueryBuilder = this.getQuery(mobileDevice);
        AggregationBuilder aggregationBuilder = AggregationBuilders.max("maxGpdate").field("gpggaDate");
        NativeSearchQueryBuilder builder = new NativeSearchQueryBuilder();
        builder.addAggregation((AbstractAggregationBuilder)aggregationBuilder);
        builder.withQuery(boolQueryBuilder);
        builder.withPageable(PageRequest.of(1, 1));
        SearchQuery query = builder.build();
//        SearchHits<Rover.OneRover> search = elasticsearchRestTemplate.search(query, Rover.OneRover.class);
        AggregatedPage<Rover.OneRover> search = elasticsearchRestTemplate.queryForPage(query, Rover.OneRover.class);
        Aggregations aggregations = search.getAggregations();
        if(aggregations==null){
            return null;
        }
        Aggregation maxGpdate = aggregations.get("maxGpdate");
        if(maxGpdate == null){
            return null;
        }
        ParsedMax parseMax =  (ParsedMax)maxGpdate;
        if(parseMax == null){
            return null;
        }

        return parseMax.getValue();
    }

    public static void main(String[] args) {
        MobileStationServiceImpl mobileStationService = new MobileStationServiceImpl();
        mobileStationService.getMobileStationMonitor(null,null,10);
    }

}
